package com.kygeng.namenode;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Title: DataNodeManager
 * @Author: kygeng
 * @Package: com.kygeng.namenode
 * @Date: 2025/3/2 18:16
 * @Description: 这个组件，就是负责管理集群里的所有的datanode的
 */
public class DataNodeManager {

    /**
     * 集群中的所有的datanode的信息
     */
    private Map<String, DataNodeInfo> datanodes =
            new ConcurrentHashMap<String, DataNodeInfo>();

    public DataNodeManager() {
        new DataNodeAliveMonitor().start();
    }

    /**
     * datanode进行注册
     *
     * @param ip
     * @param hostname
     */
    public Boolean register(String ip, String hostname) {
        DataNodeInfo datanode = new DataNodeInfo(ip, hostname);
        datanodes.put(ip + "-" + hostname, datanode);
        System.out.println("DataNode注册：ip=" + ip + ",hostname=" + hostname);
        return true;
    }

    /**
     * datanode进行心跳
     *
     * @param ip
     * @param hostname
     * @return
     */
    public Boolean heartbeat(String ip, String hostname) {
        DataNodeInfo datanode = datanodes.get(ip + "-" + hostname);
        datanode.setLatestHeartbeatTime(System.currentTimeMillis());
        System.out.println("DataNode发送心跳：ip=" + ip + ",hostname=" + hostname);
        return true;
    }

    /**
     * 用于检测datanode是否存活的线程
     */
    class DataNodeAliveMonitor extends Thread {

        @Override
        public void run() {
            try {

                while (true) {
                    //需要删除的datanode
                    List<String> toRemoveDatanode = new ArrayList<String>();

                    Iterator<DataNodeInfo> dataNodeInfoIterator = datanodes.values().iterator();
                    DataNodeInfo datanode = null;
                    while (dataNodeInfoIterator.hasNext()) {

                        datanode = dataNodeInfoIterator.next();

                        if (System.currentTimeMillis() - datanode.getLatestHeartbeatTime() > 90 * 1000) {
                            toRemoveDatanode.add(datanode.getIp() + "-" + datanode.getHostname());
                        }
                    }
                    // 移除datanode
                    if (!toRemoveDatanode.isEmpty()) {
                        for (String toRemoveDataNode : toRemoveDatanode) {
                            datanodes.remove(toRemoveDataNode);
                        }
                    }
                    Thread.sleep(30 * 1000);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
